package com.social.utils;

import cn.hutool.json.JSONUtil;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;


import javax.websocket.*;
import javax.websocket.server.PathParam;
import javax.websocket.server.ServerEndpoint;
import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;

import static com.social.constant.RabbitMqConstants.*;

/**
 * @className: WebSocket
 * @description: TODO websocket工具类
 * @author: XiaoXu
 * @date: 2022/11/26
 **/

@Component
@Slf4j
@ServerEndpoint("/websocket/{id}") //暴露的ws应用的路径
public class WebSocket {

    // 用来存储服务连接对象
    private static Map<String ,Session> clientMap = new ConcurrentHashMap<>();

    private static RabbitTemplate rabbitTemplate;

    @Autowired
    public void setRabbitTemplate(RabbitTemplate rabbitTemplate){
        this.rabbitTemplate = rabbitTemplate;
    }

    private static StringRedisTemplate stringRedisTemplate;

    @Autowired
    public void setStringRedisTemplate(StringRedisTemplate stringRedisTemplate){
        this.stringRedisTemplate = stringRedisTemplate;
    }
    public static final String DELAY_MESSAGE_ID = "delay:message:id";
    public static final String MESSAGE_SESSION_SET = "message:session:set";
    public static final String MESSAGE_SESSION_ID = "message:session:id:";

    public static final Long MESSAGE_SESSION_ID_TTL = 10L;
    /**
     * 客户端与服务端连接成功
     * @param session
     */
    @OnOpen
    public void onOpen(Session session, @PathParam("id") Long id){
        /*
            do something for onOpen
            与当前客户端连接成功时
         */
        System.out.println("连接成功userId :" + UserHolder.getUser().getId() + "id :" + id + "session:"+session);
        clientMap.put(session.getId(),session);

        stringRedisTemplate.opsForZSet().add(DELAY_MESSAGE_ID,session.getId(),System.currentTimeMillis());

    }

    /**
     * 客户端与服务端连接关闭
     * @param session
     */
    @OnClose
    public void onClose(Session session){
        /*
            do something for onClose
            与当前客户端连接关闭时
         */
        clientMap.remove(session.getId());
        stringRedisTemplate.opsForZSet().remove(DELAY_MESSAGE_ID,session.getId());
    }

    /**
     * 客户端与服务端连接异常
     * @param error
     * @param session
     */
    @OnError
    public void onError(Throwable error,Session session) {
        System.out.println("连接异常error:"+error+"session:" + session);
    }

    /**
     * 客户端向服务端发送消息
     * @param message
     * @throws IOException
     */
    @OnMessage
    public void onMsg(Session session,String message) {
        /*
            do something for onMessage
            收到来自当前客户端的消息时
         */
        stringRedisTemplate.opsForZSet().add(DELAY_MESSAGE_ID,session.getId(),System.currentTimeMillis());
        Map<String,String> map = new HashMap<>();
        map.put("message",message);
        map.put("id",session.getPathParameters().get("id").toString());

        String messageId = UUID.randomUUID().toString().replace("-", "");
        map.put("messageId",messageId);
        //sendAllMessage(session.getPathParameters().get("id").toString(),message);

        rabbitTemplate.convertAndSend(DELAY_EXCHANGE,DELAY_EMAIL_ROUNTING, map,new CorrelationData(MESSAGE_SESSION_ID+messageId));
        map.put("exchange",DELAY_EXCHANGE);
        map.put("rounting",DELAY_EMAIL_ROUNTING);
        stringRedisTemplate.opsForValue().set(MESSAGE_SESSION_ID+messageId,JSONUtil.toJsonStr(map),MESSAGE_SESSION_ID_TTL, TimeUnit.MINUTES);
        System.out.println("-------------------------------------"+map+message);
    }

    @RabbitListener(queues = DELAY_QUEUE)
    public void delayEmailHandler(Message message, Channel channel) throws IOException {
        String json = new String(message.getBody());
        Map map = JSONUtil.toBean(json, Map.class);
        String messageId = map.get("messageId").toString();
        Boolean isMember = stringRedisTemplate.opsForSet().isMember(MESSAGE_SESSION_SET, messageId);
        if(isMember){
            channel.basicReject(message.getMessageProperties().getDeliveryTag(),false);
            return;
        }

        try {
            sendAllMessage(map.get("id").toString(),map.get("message").toString());
            stringRedisTemplate.opsForSet().add(MESSAGE_SESSION_SET,messageId);
            channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
        } catch (IOException e) {
            channel.basicReject(message.getMessageProperties().getDeliveryTag(),false);
            System.out.println("异常：---"+e);
        }
    }
    //向所有客户端发送消息（广播）
    private void sendAllMessage(String id,String message){
        System.out.println("进入广播");
        Set<String> sessionIdSet = clientMap.keySet(); //获得Map的Key的集合
        // 此处相当于一个广播操作
        for (String sessionId : sessionIdSet) { //迭代Key集合
            Session session = clientMap.get(sessionId); //根据Key得到value
            if (id.equals(session.getPathParameters().get("id").toString())){
                session.getAsyncRemote().sendText(message); //发送消息给客户端
            }
        }
    }
}
